package com.jcwy.common.listener;

import cn.hutool.core.io.FileUtil;
import cn.hutool.core.thread.ThreadUtil;
import cn.hutool.core.util.ArrayUtil;
import cn.hutool.core.util.StrUtil;
import com.jcwy.common.properties.MonitorSourceProperties;
import com.jcwy.common.util.SpringContextUtil;
import com.ververica.cdc.connectors.mysql.source.MySqlSource;
import com.ververica.cdc.connectors.mysql.table.StartupOptions;
import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.RestOptions;
import org.apache.flink.runtime.state.hashmap.HashMapStateBackend;
import org.apache.flink.runtime.state.storage.FileSystemCheckpointStorage;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStreamSink;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.springframework.beans.factory.InitializingBean;

import javax.annotation.Resource;
import java.io.File;
import java.util.Arrays;
import java.util.Optional;
import java.util.Properties;

/**
 * @Author: wjs
 * @CreateTime: 2023-03-03 11:49
 */
@Slf4j
public class MysqlSourceExample implements InitializingBean {

    private MonitorSourceProperties monitorSourceProperties;
//    private final String checkPointFileAddr = "///checkpoints-data/flink-ck";
//    private final String checkPointFileTrueAddr = "file:" + checkPointFileAddr;

    @Resource
    private CustomSink customSink;

    /**
     * 获取本地中的checkpoint文件，根据创建时间进行倒序，然后拿取第一个。
     *
     * @return
     */
    private String getCheckPointAddr() {
        String checkPointFileAddr = monitorSourceProperties.getCheckPointFileAddr();
        if (FileUtil.exist(checkPointFileAddr)) {
            File[] files = FileUtil.ls(checkPointFileAddr);
            if (ArrayUtil.isNotEmpty(files)) {
                Arrays.sort(files, (o1, o2) -> {
                    long diff = o1.lastModified() - o2.lastModified();
                    if (diff > 0) {
                        return -1;
                    } else if (diff == 0) {
                        return 0;
                    } else {
                        return 1;
                    }
                });
                // 只保留最近5个文件
                if (files.length > 5) {
                    for (int i = 5; i < files.length; i++) {
                        File file = files[i];
                        String s = checkPointFileAddr + "/" + file.getName();
                        if (FileUtil.exist(s)) {
                            FileUtil.del(s);
                            log.info("------------删除了：{} 文件", s);
                        }
                    }
                }
                // 5个文件中挨个找到chk开头的文件
                for (File file : files) {
                    String fileName = file.getName();
                    String[] list = file.list();
                    Optional<String> chk = Arrays.stream(list).filter(m -> m.startsWith("chk")).findFirst();
                    if (chk.isPresent()) {
                        return fileName + "/" + chk.get();
                    }
                }
            } else {
                log.error("该文件夹为空！！！");
                return null;
            }
        } else {
            log.error("文件地址不存在！！！");
            return null;
        }
        return null;
    }

    public void run() throws Exception {
        Properties properties = new Properties();
        properties.put("decimal.handling.mode", "String");
        MySqlSource<String> source = MySqlSource.<String>builder()
                .hostname(monitorSourceProperties.getHostname())
                .port(monitorSourceProperties.getPort())
                .databaseList(monitorSourceProperties.getDatabaseList().split(","))
                // 这里如果不配置表信息的话，默认就监听aiji数据库的全部表。
                .tableList(monitorSourceProperties.getTableList().split(","))
                .username(monitorSourceProperties.getUsername())
                .password(monitorSourceProperties.getPassword())
                // 将数据库的decimal类型转换为String类型，防止获取到的decimal数据出现乱码数据。
                .debeziumProperties(properties)
                .deserializer(new JsonDebeziumDeserializationSchema())
                // 启动时从哪里开始读，这里设置的是会先进行全表快照（启动时扫描全表）
                .startupOptions(StartupOptions.initial())
                // 是否监听表结构的变化
                .includeSchemaChanges(true)
                .build();
        // 启动一个webUI，指定本地webUI端口
        Configuration configuration = new Configuration();
        configuration.setInteger(RestOptions.PORT, 8081);
        // 指定从那个快照文件进行恢复，这样就不用每次启动的时候就扫描全部的数据库数据了。生产环境中这里的文件地址是作为参数进行动态传递的，这里只是作为测试演示。
        String pointAddr = this.getCheckPointAddr();
        String checkPointFileTrueAddr = monitorSourceProperties.getCheckPointFileTrueAddr();
        if (StrUtil.isNotBlank(pointAddr)) {
            log.info("恢复的检查点路径是：{}", pointAddr);
            configuration.setString("execution.savepoint.path", checkPointFileTrueAddr + "/" + pointAddr);
        }
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(configuration);

        // 检查间隔时间点
        env.enableCheckpointing(30000);
        // checkpoint必须在一分钟内完成，否则就会被抛弃。
        env.getCheckpointConfig().setCheckpointTimeout(60000);
        // 设置模式为精确一次（默认）
        env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
        // 统一时间值允许一个checkpoint进行。
        env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);
        // 最多可容忍失败的次数
        env.getCheckpointConfig().setTolerableCheckpointFailureNumber(2);
        // 将ck记录写入到本地文件中进行保存,也可以使用HDFS的方式去保存。
        env.getCheckpointConfig().setCheckpointStorage(new FileSystemCheckpointStorage(checkPointFileTrueAddr));
//        env.getCheckpointConfig().setCheckpointStorage("");
        env.setStateBackend(new HashMapStateBackend());
        DataStreamSink<String> sink = env.fromSource(source, WatermarkStrategy.noWatermarks(), "MySQL Source")
                .addSink(customSink);
        sink.name("MySQL数据监控");
        // 启动
        env.execute();
    }

    /**
     * 类实例化完成后的操作
     *
     * @throws Exception
     */
    @Override
    public void afterPropertiesSet() {
        // 初始化配置文件类
        MonitorSourceProperties bean = SpringContextUtil.getSpringContext().getBean(MonitorSourceProperties.class);
        this.monitorSourceProperties = bean;
        // 这里尽量使用异步线程去执行，不然后续操作中如果有context.getBean()调用，会发生死锁。
        ThreadUtil.execAsync(() -> {
            try {
                run();
            } catch (Exception e) {
                throw new RuntimeException(e);
            }
        });
    }
}
